

package v2store

import (
	"container/list"
	"path"
	"strings"
	"sync"
	"sync/atomic"

	"go.etcd.io/etcd/server/v3/etcdserver/api/v2error"
)

// 一个watcherHub包含所有订阅的watcherHub 
// watchers是一个映射，watcherpath作为键，watcher作为值
// EventHistory为watcherHub保留旧事件。它用于帮助
// 观察者获得连续的事件历史记录。或者，监视程序可能会错过在第一个监视命令的结束和第二个命令的开始之间发生的
// 事件。
type watcherHub struct {
	count int64 // 当前watcherHub实例中保存的watcher实例个数。
	mutex        sync.Mutex
	// 维护节点与监听该节点的watcher实例的对应关系，
	// 其中key是node.Path字段值，value则是监听该节点的watcher实例列表。
	watchers     map[string]*list.List
	// 保存最近发生的修改操作对应的Event实例，当EventHistory的容量达到上限之后，
	// 继续向其中添加Event实例，则会自动删除最早添加的Event实例。
	EventHistory *EventHistory
}

// newWatchHub创建一个watcherHub。容量决定了我们将
// 在事件历史记录中保留多少事件。
// 通常，我们只需要保留一小部分历史记录[小于20K]。
// 理想情况下，它应该小于20K/s[最大吞吐量]*2*50ms[RTT]=2000 
func newWatchHub(capacity int) *watcherHub {
	return &watcherHub{
		watchers:     make(map[string]*list.List),
		EventHistory: newEventHistory(capacity),
	}
}

// 当客户端请求etcd服务端为某个指定的节点添加watcher时，
// 是通过调用watcherHub.watch() 方法完成的。在添加watcher之前，
// 会先查找EventHistory确定sincelndex～Currentlndex之间是否发生了触发待添加watcher的操作。
func (wh *watcherHub) watch(key string, recursive, stream bool, index, storeIndex uint64) (Watcher, *v2error.Error) {
	reportWatchRequest()
	// 在EventHistory中，从index开始查找对应的Event,
	event, err := wh.EventHistory.scan(key, recursive, index)

	if err != nil {
		err.Index = storeIndex
		return nil, err
	}

	w := &watcher{
		eventChan:  make(chan *Event, 100), // 使用缓冲通道
		recursive:  recursive,	// 根据recursive参数设置是否监听指定节点的子节点
		stream:     stream,
		sinceIndex: index,	// 根据index参数设置从哪个事件开始监听
		startIndex: storeIndex,	// 创建该watcher实例时对应的CurrentIndex值
		hub:        wh,	// 关联的watcherHub实例
	}
	// 相关的加锁和解锁操作
	wh.mutex.Lock()
	defer wh.mutex.Unlock()
	// 如果在EventHistory中存在合适的Event，
	// 则表示从sinceIndex到Currentindex为止,发生了触友该watcher的操作
	if event != nil {
		ne := event.Clone()
		ne.EtcdIndex = storeIndex
		// 将查找到的Event写入eventChan通道中等待处理
		w.eventChan <- ne
		return w, nil	// 直接返回watcher
	}
	// 如果未查找到合适的Event,则表示从sinceIndex到Currentindex为止，
	// 并没有发生过触发该watcher的操作
	l, ok := wh.watchers[key]	// 查找节点对应的watcher列表

	var elem *list.Element

	if ok { // 将当前watcher添加到列表尾部
		elem = l.PushBack(w)
	} else { // 第一次为节点添加watcher时，会先创建对应的watcher列表
		l = list.New()
		elem = l.PushBack(w)
		wh.watchers[key] = l
	}

	w.remove = func() {
		if w.removed { // 检测当前watcher实例是否已经被删除
			return
		}
		w.removed = true	// 标记当前watcher已被删除
		l.Remove(elem)	// 将当前watcher从watcherHub中删除
		atomic.AddInt64(&wh.count, -1)	// 更新watcherHub中保存的watcher个数
		reportWatcherRemoved()
		if l.Len() == 0 {	// 如果没有任何监听某节点，则将该节点对应的watcher 列表删除
			delete(wh.watchers, key)
		}
	}
	// 更新当前watcherHub中记录的watcher数量
	atomic.AddInt64(&wh.count, 1)
	reportWatcherAdded()

	return w, nil
}

func (wh *watcherHub) add(e *Event) {
	wh.EventHistory.addEvent(e)
}

// 发生修改操作时对应的处理逻辑。
// storage会通过watcherHub.notify（）方法触发监昕对应节点的watcher
func (wh *watcherHub) notify(e *Event) {
	// 首先将修改操作对应的Event实例保存到EventHistory中
	e = wh.EventHistory.addEvent(e) // 将事件添加到事件历史中
	segments := strings.Split(e.Node.Key, "/")	// 按照“／”切分发生修改的节点的路径
	currPath := "/"

	// 遍历切分结果，并调用notifyWatchers（）方法，通知当前节点和所有父节点上的watcher,
	// 这里举个例子，如果“／foo/bar”节点被修改了，那么不仅监听“／foo/bar＂节点的watcher会被触发，
	// 监听“／fo。”节点的watcher也可能被触发
	for _, segment := range segments {
		currPath = path.Join(currPath, segment)
		// 通知对当前路径更改感兴趣的观察者
		wh.notifyWatchers(e, currPath, false)
	}
}

// watcher.notifyWatchers（）方法，该方法会查找节点对应的watcher列表，
// 并逐一触发列表中的watcher实例，当watcher实例（streamwatcher除外）被成功触发之后，则会将其清除，
func (wh *watcherHub) notifyWatchers(e *Event, nodePath string, deleted bool) {
	wh.mutex.Lock()
	defer wh.mutex.Unlock()
	// 获取监听指定节点的watcher列表
	l, ok := wh.watchers[nodePath]
	if ok {
		curr := l.Front()
		for curr != nil {	// 遍历该watcher列表
			next := curr.Next() // 保存对列表中下一个路径的引用
			w, _ := curr.Value.(*watcher)	// 获取watcher实例
			originalPath := e.Node.Key == nodePath
			// 调用watcher.notify（）方法，触发相应的watcher
			if (originalPath || !isHidden(nodePath, e.Node.Key)) && w.notify(e, originalPath, deleted) {
				// 检测是否为streamwatcher，如果不是，则需要将该watcher清除
				if !w.stream {
					w.removed = true
					l.Remove(curr)	// 从当前watcher列表中删除
					atomic.AddInt64(&wh.count, -1)	// 更新count字段
					reportWatcherRemoved()
				}
			}

			curr = next // 继续处理watcher列表中的下一个watcher实例
		}
		// 如果列表中的watcher实例全部被清空，则将该列表从watchHub中清除
		if l.Len() == 0 {
			delete(wh.watchers, nodePath)
		}
	}
}

// 克隆函数克隆观察者，并返回克隆的观察者。
// 仅克隆静态内容。不要克隆当前的观察者。
func (wh *watcherHub) clone() *watcherHub {
	clonedHistory := wh.EventHistory.clone()

	return &watcherHub{
		EventHistory: clonedHistory,
	}
}

// isHidden检查是否将密钥路径视为隐藏到监视路径，例如，
// 最后一个元素被隐藏或位于隐藏目录
func isHidden(watchPath, keyPath string) bool {
	// 删除目录时，监视路径可能比实际密钥路径
	// 更深，删除/foo时，我们还需要在/foo/bar上通知观察者。
	if len(watchPath) > len(keyPath) {
		return false
	}
	// 如果watch path仅仅是一个“/”，则路径开始后将不带“/”
	// 添加一个“/”，以处理watchPath为“/”时的特殊情况
	afterPath := path.Clean("/" + keyPath[len(watchPath):])
	return strings.Contains(afterPath, "/_")
}
